
#include "common.h"
#include "../util/util.hpp"
#include "../util/poll_set.hpp"                                                                                                                                                           
#include "../util/poll_manager.hpp"
#include "../transport/transport.hpp"
#include "../transport/transport_tcp.hpp"

#include <queue>
#include <utility> // pair
#include <thread>
#include <condition_variable>
#include <mutex>

#include <stdlib.h>
#include <string>
#include <string.h>

#include <boost/make_shared.hpp>
#include <boost/smart_ptr.hpp>
#include <functional>
#include <assert.h>
#include <memory>
#include <unistd.h>


#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>        
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <string.h>    

#include <time.h>
 
#define handle_error(msg) \
	do { perror(msg); exit(EXIT_FAILURE); } while(0)


FILE *fp=NULL;
int tcp_fd=-1;
int cfd=-1;
int addrlen=0;
int size_number=0;
ssize_t recv_size=0;	
//char recv_buf[10240+1]={0};
//char recv_buf[1024+1]={0};

int write_sum=0;
ssize_t write_size=0;	
long unsigned int file_size = 0;
long unsigned int total_recv_byte=0;

std::queue<std::string> recv_queue;
//std::queue<std::pair<uint8_t*, int>> recv_queue;
//std::queue<std::pair<char*, int>> recv_queue;
std::mutex cv_mutex;
std::condition_variable cv_queue;

TransportTCPPtr tcp_transport = NULL;
TransportTCPPtr tcp_transport2 = NULL;



void OnTCPReadable(const TransportPtr& transport);
void readTCPTransport(const TransportPtr& transport);

void OnTCPReadable(const TransportPtr& transport)
{
  (void)transport;
  //assert(udp_transport == transport);

  readTCPTransport(transport);
}


void readTCPTransport(const TransportPtr& transport)
{
  bool reading = true;
  bool has_read_callback = true;

  int size = 1024;
  boost::shared_array<uint8_t> read_buffer;
  read_buffer = boost::shared_array<uint8_t>(new uint8_t[size]);
  //char read_buffer[size];

  uint32_t read_size=size;

  uint32_t read_filled=0;

  while(has_read_callback)
  {
    assert(read_buffer);
    uint32_t to_read = read_size - read_filled;
    if(to_read > 0)
    {
      int32_t bytes_read = transport->read(read_buffer.get()+read_filled, to_read);
      //int32_t bytes_read = transport->read((unsigned char*)read_buffer+read_filled, to_read);
      //ROS_DEBUG_NAMED("superdebug", "Server connection read %d bytes", bytes_read);
      TRACE("[INFO] superdebug Server connection read %d bytes", bytes_read);
      if(bytes_read<0)
      {
        TRACE("[ERROR] Bad read, throw away results and report error");
        read_buffer.reset();
        read_size = 0;
        read_filled = 0;
        has_read_callback = false;
        break;
      }
      //read ok

//TRACE("[INFO] read_buffer:%s", read_buffer.get());
      read_filled += bytes_read;
total_recv_byte+=bytes_read;

TRACE("just after transport->read()");
TRACE("[TOTAL SIZE] %ld", total_recv_byte);
TRACE("[FILE SIZE] %ld", file_size);


    }
    assert((int32_t)read_size>=0);
    assert((int32_t)read_filled>=0);
    //assert(read_filled<=read_size, "read_filled=%d, read_size=%d", read_filled, read_size);
    if(read_filled>read_size)
    {
        TRACE("[ASSERT] read_filled=%d, read_size=%d", read_filled, read_size);
        assert(0);
    }

    if(read_filled < read_size || read_filled==read_size || total_recv_byte >= file_size)
    {
        //------------push queue------------------------
        //uint8_t* buff_tmp_read = new uint8_t[to_read+1];
        //memset(buff_tmp_read, 0, sizeof(buff_tmp_read));
        ////char* buff_tmp_read = new char[to_read];
        //memcpy(buff_tmp_read, read_buffer.get()+read_filled, to_read);
        ////strcpy(buff_tmp_read, read_buffer+read_filled);
#ifdef _DEBUG
        printf("----------------------------------------------------------\n");
        printf("[INFO] buff_len:\n%d\n", read_size);
        printf("[INFO] buff_read:\n%s\n", read_buffer.get());
        printf("[INFO] int of end char:\n%d\n", read_buffer.get()[read_filled-1]);
        printf("[INFO] char of end char:\n%c\n", read_buffer.get()[read_filled-1]);
        printf("----------------------------------------------------------\n");
#endif
        //TRACE("just before push queue");
        TRACE("[TOTAL SIZE] %ld", total_recv_byte);
        TRACE("[FILE SIZE] %ld", file_size);
        //recv_queue.push(std::pair<uint8_t*, int>(buff_tmp_read, to_read));
        //recv_queue.push(std::pair<uint8_t*, int>(buff_tmp_read, to_read));
        
        //std::string tmpstr((char*)read_buffer.get());
        //int tmplen = tmpstr.size();
        //if(tmpstr[tmplen-1]!='\n')
        //  recv_queue.push(tmpstr.substr(0, tmpstr.size()-1));
        //else 
        //  recv_queue.push(tmpstr);
        read_buffer.get()[read_filled]='\0';
        recv_queue.push(string((char*)read_buffer.get()));

        if(recv_queue.size()>=1){
          std::unique_lock<std::mutex> lk(cv_mutex);
          cv_queue.notify_all();
        }
        //------------push queue------------------------
        
        TRACE("[INFO] Finish read, store off");
        read_buffer.reset();
        read_size = 0;
        read_filled = 0;
        has_read_callback = false;
    }
    else
    {
      break;
    }
  }
  if(!has_read_callback)
  {
    //transport->disableRead();
  }
  reading=false;
}



void checkForShutdown()
{}
void acceptCallback(const TransportTCPPtr& transport)
{
  INFO("acceptCallback!");
  tcp_transport2 = transport;
}


int
main(int argc, const char *argv[]){
  /*------------------------------------------------------------*/
	
	//remote_addr.sin_addr.s_addr=inet_addr(argv[2]);
	//remote_addr.sin_port=htons(atoi(argv[3]));
	
	if(argc!=2){		
		fprintf(stderr,"Usage: %s <save_filename> \n",argv[0]);
		exit(EXIT_FAILURE);		
	}

  printf("file name to recv: %s\n", argv[1]);
	
	//if((fp = fopen(argv[1], "w+")) == NULL){
	if((fp = fopen(argv[1], "wb")) == NULL){
		handle_error("fopen");		
	}
	
  /*------------------------------------------------------------*/



  /*------------------------------------------------------------*/
  int servfd, clientfd;
  struct sockaddr_in serv_addr, client_addr;
  memset(&serv_addr, 0, sizeof(serv_addr));
  memset(&client_addr, 0, sizeof(client_addr));

  if((servfd = socket(AF_INET, SOCK_STREAM, 0))==-1)
  {
    INFO("create tcp socket failed: ", strerror(errno));
    return 0;
  }
  serv_addr.sin_family = AF_INET;
  serv_addr.sin_port = htons(6060);
  serv_addr.sin_addr.s_addr = inet_addr("127.0.0.1");
  if(bind (servfd, (sockaddr*)&serv_addr, sizeof(serv_addr))==-1)
  {
    INFO("bind failed : ", strerror(errno));
    return 0;
  }
  if(listen(servfd, 1024)==-1)
  {
    INFO("listen failed : ", strerror(errno));
    return 0;
  }
  socklen_t client_size = 0;
  if((clientfd = accept(servfd, (sockaddr*)&client_addr, &client_size))==-1)
  {
    INFO("accept failed : ", strerror(errno));
    return 0;
  }
  char tcpbuf[1024];
  char tcpsd[1024];
  memset(tcpbuf, 0, sizeof(tcpbuf));
  memset(tcpsd, 0, sizeof(tcpsd));
  int rdstate = read(clientfd, tcpbuf, sizeof(tcpbuf));
  if(rdstate==0)
  {
    std::cout<<"client exit!"<<std::endl;
    return 0;
  }
  if(strncasecmp(tcpbuf, "SIZE:", 5)!=0)
  {
    std::cout<<"SIZE cmd wrong!"<<std::endl;
    return 0;
  }
  for(int i = 0; i < 5; i++) tcpbuf[i] = 0;
  strcpy(tcpbuf, tcpbuf+5);
  file_size = atoi(tcpbuf);
  std::cout<<"SIZE : "<<file_size<<" Bytes"<<std::endl;

  sprintf(tcpsd, "OK");
  write(clientfd, tcpsd, strlen(tcpsd));
  
  INFO("[INFO] 已经确认FILE SIZE,开始创建TCPROS");

  //memset(tcpbuf, 0, sizeof(tcpbuf));
  shutdown(servfd, SHUT_RDWR);
  shutdown(clientfd, SHUT_RDWR);

  /*------------------------------------------------------------*/
  std::thread thread_write_file([&]{
      bool is_finished = false;
      //uint8_t* buf_recv = NULL;
      //char* buf_recv = NULL;
      while(1)
      {
        if(recv_queue.empty()){
          std::unique_lock<std::mutex> lk(cv_mutex);
          cv_queue.wait(lk, [&]{return !recv_queue.empty();});

          continue; 
        }

        while(!recv_queue.empty())
        {
          //------------------pop buf start-----------------------
          //int buf_len = recv_queue.front().second;
          int buf_len = recv_queue.front().size();
          char buf_recv[buf_len];
          //buf_recv = recv_queue.front().first;
          strcpy(buf_recv, recv_queue.front().c_str());
          recv_queue.pop();
          //------------------pop buf start-----------------------
///printf("----------------------------------------------------------\n");
///std::cout<<"[BUF_LEN]"<<std::endl;
///std::cout<<buf_len<<std::endl;
///std::cout<<"[BUF_RECV]"<<std::endl;
///std::cout<<buf_recv<<std::endl;
///printf("----------------------------------------------------------\n");

          //------------------write file start-----------------------
		      if(fwrite(buf_recv, buf_len, 1, fp)<0){
		      	fclose(fp);
		      	fprintf(stderr,"Error %s file!\n", argv[1]);
		      	exit(EXIT_FAILURE);
		      }
          fflush(fp);
          //------------------write file end-----------------------
          //------------------delete tmp buffer-----------------------
          //delete [] buf_recv;
          //buf_recv = NULL;
          //------------------delete tmp buffer-----------------------
          
          write_sum += buf_len;
          procBar(write_sum, file_size);
          if(write_sum >= file_size){is_finished=true; break;}
        }
        if(is_finished==true)break;
      }
      TRACE("[INFO]write ok!\n");
      fflush(stdout);
    });
  /*------------------------------------------------------------*/

  /*------------------------------------------------------------*/
  INFO("[INFO] start PollManager...loading...");

  PollManager::instance()->addPollThreadListener(checkForShutdown);
  PollManager::instance()->start();

  INFO("[INFO] start PollManager...ok!");


  INFO("[INFO] create tcp_transport...");
  tcp_transport= std::make_shared<TransportTCP>(&PollManager::instance()->getPollSet());
  INFO("[INFO] ok!");



  INFO("[INFO] connect...");
  //if(!tcp_transport->connect("127.0.0.1", 9090, "127.0.0.1", 8080, false))
  
  //if(!tcp_transport->createIncoming(9090, true))
  if(!tcp_transport->listen(9090, 100, acceptCallback))
  {
    INFO("[FATAL] KCPServer listen failed!");
    INFO("[BREAK]");assert(0);// ROS_BREAK(); 
  }
  INFO("[INFO] TCPServer connect successfully!");

  //do{}while(tcp_transport2==NULL);
  INFO("[INFO] wait for accept");
  while(tcp_transport2==NULL){usleep(1);}
  INFO("[INFO] accept to tcp_transport2!");
  /*------------------------------------------------------------*/
  //#define CLOCK1
  #define CLOCK2
  
  #ifdef CLOCK1
    int begintime, endtime;
    begintime = clock();
  #endif
  #ifdef CLOCK2
    time_t begintime, endtime;
    //begintime = time(NULL);
    begintime = iclock64();
  #endif
  /*------------------------------------------------------------*/

  //-------------------------------------------------------
  tcp_transport2->setReadCallback(OnTCPReadable);
  tcp_transport2->enableRead();
  readTCPTransport(tcp_transport2);
  //while(true){usleep(1000000);}
  //-------------------------------------------------------



	while(1){
										
    //INFO("[INFO] start read..");
    //readTransport(tcp_transport2, (unsigned char*)recv_buf, sizeof(recv_buf)-1);
    
    
		//if((recv_size=tcp_transport2->read((unsigned char*)recv_buf, sizeof(recv_buf)-1))==-1){
    //  INFO("[INFO]read bytes < 0 , wrong, return !");
    //  continue;
		//	//fclose(fp);
	  //  //handle_error("recv");	
		//}
    ////INFO("[INFO] finish a read..");


    //------------push queue------------------------
    //recv_queue.push(string(recv_buf));

    //if(recv_queue.size()==1){
    //  std::unique_lock<std::mutex> lk(cv_mutex);
    //  cv_queue.notify_all();
    //}
    //------------push queue------------------------

		//bzero(recv_buf, sizeof(recv_buf));

		//size_number=size_number+recv_size;
    //procBar(size_number, file_size);
      if( write_sum >= file_size )break;
		  usleep(1000000);
	}

  /*------------------------------------------------------------*/
  thread_write_file.join();
  /*------------------------------------------------------------*/
  INFO("recv and write ok!");
  fflush(stdout);

#ifdef CLOCK1
  endtime = clock();
  double proc_time = (double)(endtime - begintime)/(double)CLOCKS_PER_SEC;//sec
#endif
#ifdef CLOCK2
  //endtime = time(NULL);
  //double proc_time =(double)(difftime(endtime , begintime));//sec
  endtime = iclock64();
  double proc_time =(double)difftime(endtime , begintime)/1000.0;//sec
#endif
  double size_Mb = (((double)file_size*8)/(1024.0*1024.0));//Mbit
  double rate = size_Mb/proc_time;//Mbit/s
  printf("|:::::::::::::::::::::::::::|\n");
  printf("file size: %lf Mb\n", size_Mb);
  printf("use time: %lf s\n",proc_time);
  printf("rate: %lf Mbps\n", rate );                                                                                                           
  printf("|:::::::::::::::::::::::::::|\n");


  printf("enter while(1)...\n");
  fflush(stdout);


  while(1){};

  INFO("[INFO] shutdown PollManager...loading...");

  PollManager::instance()->shutdown();

  INFO("[INFO] shutdown PollManager...ok!");

	
	fclose(fp);
  /*------------------------------------------------------------*/

  close(servfd);
  close(clientfd);

	return 0;
}
